// Copyright 2014 The mqrouter Author. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package xredismq

import (
	"context"
	"encoding/json"
	"fmt"
	"gitee.com/go-mid/infra/xcache/xredis"
	"gitee.com/go-mid/infra/xmq/mqconf"
	"gitee.com/go-mid/infra/xtime"
	"github.com/go-redis/redis"
	"strings"
)

type ReaderConfig struct {
	mqconf.MQConfig
	RedisConf xredis.RedisConfig `json:"redis_conf"`
	Topic     string             `json:"topic"`
	Group     string             `json:"group"`
	//消费者唯一标识,可以是服务实例id
	Consumer string       `json:"consumer"`
	Model    RedisMQModel `json:"model"`
}
type RedisMQModel int

const (
	RedisMQModel_PubSub RedisMQModel = 0
	RedisMQModel_Stream RedisMQModel = 1
)

type RedisReader struct {
	redisClient *redis.Client
	config      *ReaderConfig
	stop        chan struct{}
	//pub/sub模式下使用
	sub *redis.PubSub
}

func NewRedisReader(readerConfig *ReaderConfig) (*RedisReader, error) {
	var ctx = context.Background()
	redisClient, err := xredis.NewRedisClient(ctx, &readerConfig.RedisConf)
	if err != nil {
		return nil, err
	}
	client := redisClient.GetRawClient(ctx)
	redisReader := &RedisReader{
		redisClient: client,
		stop:        make(chan struct{}),
		config:      readerConfig,
	}
	switch readerConfig.Model {
	case RedisMQModel_PubSub:
		redisReader.sub = client.Subscribe(readerConfig.Topic)
	case RedisMQModel_Stream:
		_, err = client.XGroupCreateMkStream(readerConfig.Topic, readerConfig.Group, "$").Result()
		if err != nil && !strings.Contains(err.Error(), "BUSYGROUP") {
			return nil, err
		}
	}
	return redisReader, nil
}

func (m *RedisReader) ReadMsg(ctx context.Context, v interface{}) error {
	st := xtime.NewTimeStat()
	if m.config.Model == RedisMQModel_PubSub {
		select {
		case msg := <-m.sub.Channel():
			// 成功接收消息
			mqconf.StatReqDuration(ctx, m.config.Topic, "RedisReader.ReadMessage", mqconf.TraceMessageBusTypeRedis, st.Millisecond())
			err := json.Unmarshal([]byte(msg.Payload), v)
			if err != nil {
				return err
			}
			return nil
		case <-ctx.Done():
			return fmt.Errorf("context canceled")
		}
	}
	entries, err := m.redisClient.XReadGroup(&redis.XReadGroupArgs{
		Group:    m.config.Group,
		Consumer: m.config.Consumer,
		Streams:  []string{m.config.Topic, ">"},
		//不需要ack
		NoAck: true,
	}).Result()
	mqconf.StatReqDuration(ctx, m.config.Topic, "RedisReader.ReadMessage", mqconf.TraceMessageBusTypeRedis, st.Millisecond())
	if err != nil {
		return err
	}
	for _, entry := range entries {
		for _, message := range entry.Messages {
			if msg, ok := message.Values[xaddkey]; ok && msg != "" {
				if msgStr, ok := msg.(string); ok {
					err = json.Unmarshal([]byte(msgStr), v)
					if err != nil {
						return err
					}
					return nil
				} else {
					return fmt.Errorf("msg not string")
				}
			} else {
				return fmt.Errorf("no msg")
			}
		}
	}
	return nil
}

func (m *RedisReader) FetchMsg(ctx context.Context, v interface{}) (mqconf.AckHandler, error) {

	st := xtime.NewTimeStat()
	if m.config.Model == RedisMQModel_PubSub {
		select {
		case msg := <-m.sub.Channel():
			// 成功接收消息
			mqconf.StatReqDuration(ctx, m.config.Topic, "RedisReader.FetchMsg", mqconf.TraceMessageBusTypeRedis, st.Millisecond())
			err := json.Unmarshal([]byte(msg.Payload), v)
			if err != nil {
				return &mqconf.EmptyAckHandler{}, err
			}
			return &mqconf.EmptyAckHandler{}, nil
		case <-ctx.Done():
			return &mqconf.EmptyAckHandler{}, fmt.Errorf("context canceled")
		}
	}

	entries, err := m.redisClient.XReadGroup(&redis.XReadGroupArgs{
		Group:    m.config.Group,
		Consumer: m.config.Consumer,
		Streams:  []string{m.config.Topic, ">"},
		//需要ack
		NoAck: false,
	}).Result()

	mqconf.StatReqDuration(ctx, m.config.Topic, "RedisReader.FetchMsg", mqconf.TraceMessageBusTypeRedis, st.Millisecond())
	if err != nil {
		return nil, err
	}
	for _, entry := range entries {
		for _, message := range entry.Messages {
			if msg, ok := message.Values[xaddkey]; ok && msg != "" {
				if msgStr, ok := msg.(string); ok {
					err = json.Unmarshal([]byte(msgStr), v)
					if err != nil {
						return nil, err
					}

					return &RedisMQHandler{
						msg:    message,
						client: m.redisClient,
						group:  m.config.Group,
						topic:  m.config.Topic,
					}, nil
				} else {
					return nil, fmt.Errorf("msg not string")
				}
			} else {
				return nil, fmt.Errorf("no msg")
			}
		}
	}
	return &RedisMQHandler{}, nil
}

func (m *RedisReader) Close(ctx context.Context) error {
	m.stop <- struct{}{}
	if m.sub != nil {
		m.sub.Close()
	}
	return m.redisClient.Close()
}

type RedisMQHandler struct {
	msg    redis.XMessage
	client *redis.Client
	group  string
	topic  string
}

func (m *RedisMQHandler) Ack(ctx context.Context) error {
	// 确认消息已被处理
	_, err := m.client.XAck(m.topic, m.group, m.msg.ID).Result()
	return err
}
